/*
 * Copyright 2011-2012 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.sabayframework.cluster.impl;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.sabayframework.cluster.NodeId;
import org.sabayframework.cluster.NodeMapping;
import org.sabayframework.util.Hash;


/**
 * Mapping implemented as a consistent hash.
 * @see http://weblogs.java.net/blog/tomwhite/archive/2007/11/consistent_hash.html
 * @author Alex Antonau
 * */
public final class ConsistentHashing implements NodeMapping {

	private final SortedMap<Integer, NodeId> circle;
	private final int numberOfReplicas;
	private final ReentrantReadWriteLock lock;
	
	public ConsistentHashing(int numberOfReplicas) {
		this.circle = new TreeMap<Integer, NodeId>();
		this.numberOfReplicas = numberOfReplicas;
		this.lock = new ReentrantReadWriteLock();
	}
	
	public void add(NodeId nodeId) {
		lock.writeLock().lock();
		try{
			for (int i = 0; i < numberOfReplicas; i++) {
				circle.put(Hash.intHash(nodeId.hashCode() + i), nodeId);
			}
		} finally{
			lock.writeLock().unlock();
		}
	}

	public void remove(NodeId nodeId) {
		lock.writeLock().lock();
		try{
			for (int i = 0; i < numberOfReplicas; i++) {
				circle.remove(Hash.intHash(nodeId.hashCode() + i));
			}
		} finally{
			lock.writeLock().unlock();
		}
	}
	
	public boolean contains(NodeId nodeId){
		int hash=Hash.intHash(nodeId.hashCode());
		lock.readLock().lock();
		try{
			return circle.containsKey(hash);
		} finally{
			lock.readLock().unlock();
		}
	}
	
	public NodeId get(int key) {
		int hash = Hash.intHash(key);
		lock.readLock().lock();
		try{
			if (circle.isEmpty()) {
				return null;
			}
			if (!circle.containsKey(hash)) {
				SortedMap<Integer, NodeId> tailMap = circle.tailMap(hash);
				hash = tailMap.isEmpty() ? circle.firstKey() : tailMap.firstKey();
			}
			return circle.get(hash);
		} finally{
			lock.readLock().unlock();
		}
	}

	public NodeId[] nearest(int key, int n) {
		ArrayList<NodeId> list=new ArrayList<NodeId>();
		lock.readLock().lock();
		try{
			if (circle.isEmpty()) {
				return null;
			}
			int size = Math.min(circle.size() / numberOfReplicas, n);
			if(size == 0) size=1;
			Iterator<NodeId> it = iterate(key);
			int counter = circle.size();
			while(list.size()<size && counter-->0){
				NodeId nid=it.next();
				if(!list.contains(nid))list.add(nid);
			}
		} finally{
			lock.readLock().unlock();
		}
		return list.toArray(new NodeId[list.size()]);
	}
	private Iterator<NodeId> iterate(int key) {
		final int hash = Hash.intHash(key);
		return new Iterator<NodeId>() {
			private Iterator<NodeId> innerIterator = circle.tailMap(hash).values().iterator();
			public boolean hasNext() {
				return true;
			}
			public NodeId next() {
				if(!innerIterator.hasNext())
					innerIterator = circle.values().iterator();
				return innerIterator.next();
			}
			public void remove() {	}
		};
	}

}
